package com.example.demo.Gmail.Excutor;

import com.example.demo.Gmail.Handle.WatchHandler;
import com.example.demo.Gmail.Handle.proxyHandler;
import com.example.demo.Gmail.service.BackendService;
import com.example.demo.Gmail.service.UserService;
import com.google.api.gax.core.ExecutorProvider;
import com.google.api.gax.core.InstantiatingExecutorProvider;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.io.IOException;
import java.security.GeneralSecurityException;
import java.util.concurrent.ExecutionException;


public class pullThead implements Runnable {
    WatchHandler watchHandler=new WatchHandler();
    UserService userService;
    BackendService backendService;
    public pullThead(UserService userService,BackendService backendService){
        this.backendService=backendService;
        this.userService=userService;
    }
    @Override
    public void run() {
        //大陆环境需要加代理
        proxyHandler.proxy();
        String projectId = "myproject-221302";
        String subscriptionId = "cne-sub";
        ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(projectId, subscriptionId);
// Instantiate an asynchronous message receiver
        MessageReceiver receiver =
                new MessageReceiver() {
                    @Override
                    public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
                        // handle incoming message, then ack/nack the received message
                        try {
                            watchHandler.handler(message,userService,backendService);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (ExecutionException e) {
                            e.printStackTrace();
                        } catch (GeneralSecurityException e) {
                            e.printStackTrace();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        consumer.ack();
                    }
                };
        ExecutorProvider executorProvider =
                InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(1).build();

        Subscriber subscriber =
                Subscriber.newBuilder(subscriptionName, receiver)
                        .setExecutorProvider(executorProvider)
                        .build();
        try {
            // Create a subscriber for "my-subscription-id" bound to the message receiver
            subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
            subscriber.startAsync();
            // ...
        } finally {
            // stop receiving messages
            if (subscriber == null) {
                subscriber.stopAsync();
            }
        }
    }
}
